package com.cgrs572.shortlink.project.config;

import com.cgrs572.shortlink.project.mq.consumer.ShortLinkStatsSaveConsumer;
import lombok.RequiredArgsConstructor;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static com.cgrs572.shortlink.project.common.constant.RedisKeyConstant.SHORT_LINK_STATS_STREAM_GROUP_KEY;
import static com.cgrs572.shortlink.project.common.constant.RedisKeyConstant.SHORT_LINK_STATS_STREAM_TOPIC_KEY;

/**
 * Redis Stream 消息队列配置
 */
@Configuration
@RequiredArgsConstructor
public class RedisStreamConfiguration {

    private final RedisConnectionFactory redisConnectionFactory; // Redis连接工厂对象
    private final ShortLinkStatsSaveConsumer shortLinkStatsSaveConsumer; // Redis消费者

    /**
     * 线程池对象的Bean
     * @return
     */
    @Bean
    public ExecutorService asyncStreamConsumer() {
        AtomicInteger index = new AtomicInteger(); //创建一个Integer原子类
        int processors = Runtime.getRuntime().availableProcessors(); // 当前 JVM 可用的 CPU 核心数
        // 返回一个线程池对象
        return new ThreadPoolExecutor(
                processors, // 核心线程数
                processors + processors >> 1, // 动态计算线程池的最大线程数.processors >> 1:右移一位相当于除以2，即 processors + (processors / 2)
                60, //救急线程的生存时间
                TimeUnit.SECONDS, //救急线程的生存时间单位
                new LinkedBlockingQueue<>(), // 工作队列
                // 线程工厂---通过创建runnable接口的匿名实现类来创建线程工厂中的线程
                runnable -> {
                    Thread thread = new Thread(runnable); // 创建线程对象
                    thread.setName("stream_consumer_short-link_stats_" + index.incrementAndGet()); // 指定当前线程的名称 --- incrementAndGet():原子性地Integer原子类中的Integer值加 1，并返回新值
                    thread.setDaemon(true); // 设置线程为守护线程
                    return thread;
                }
        );
    }

    /**
     *  Redis Stream 消息监听对象Bean
     * @param asyncStreamConsumer 异步执行消息消费任务的线程池
     * @return
     */
    @Bean(initMethod = "start", destroyMethod = "stop") // 指定该Bean的初始化方法start()和销毁方法stop()
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(ExecutorService asyncStreamConsumer) {
        // 步骤一：配置Redis Stream 消息监听容器的监听选项
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        .batchSize(10)// 一次最多获取多少条消息---即每次从 Stream 读取的最大消息数
                        .executor(asyncStreamConsumer) // 执行从 Stream 拉取到消息的任务流程---即异步处理消息的线程池，避免阻塞 Redis 连接。
                        // 如果没有拉取到消息，需要阻塞的时间。不能大于 ${spring.data.redis.timeout}，否则会超时
                        .pollTimeout(Duration.ofSeconds(3)) // 没有消息时，消费者阻塞等待的时间（需小于 Redis 超时时间，否则会断开）。
                        .build();

        // 步骤二：创建监听容器
            // 参数一：自动注入的 Redis 连接工厂            参数二：// 上述配置的监听选项
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer =
                StreamMessageListenerContainer.create(redisConnectionFactory, options);

        // 步骤三：订阅消息并绑定消费者
        streamMessageListenerContainer.receiveAutoAck(
                // Redis Stream 消费者组 API 中的核心方法，用于定义一个消费者（Consumer）并绑定到指定的消费者组（Consumer Group）,支持消费组竞争消费
                Consumer.from(SHORT_LINK_STATS_STREAM_GROUP_KEY, "stats-consumer"), // 将消费者"stats-consumer"绑定到消费者组group(即'short_link:stats-stream:only-group')
                // Redis Stream 消费者 API 中的核心方法，用于指定从Redis Stream 中消费者组的哪个位置开始消费。它定义了消息的读取起点（偏移量）
                    // 参数一：指定要建通的redis stream
                    // 参数二：指定从哪里开始消费
                    // 参数三：一个 消息处理器，负责消费 Stream 中的消息并执行业务逻辑
                StreamOffset.create(SHORT_LINK_STATS_STREAM_TOPIC_KEY, ReadOffset.lastConsumed()), shortLinkStatsSaveConsumer);
        return streamMessageListenerContainer;
    }
}
